package com.youshang520i.springcommonmiddleware.service.impl;

import com.youshang520i.springcommonmiddleware.service.BroadcastProducerService;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * 广播模式实现类
 *
 * @author youshang
 */
@Service
public class BroadcastProducerServiceImpl implements BroadcastProducerService {

    @Value("rocketmq.name-server")
    private String rmqNamesrvAddr;

    /**
     * 发送广播模式MQ
     *
     * @return
     */
    @Override
    public SendResult sendBroadcast() throws UnsupportedEncodingException, InterruptedException, RemotingException, MQClientException, MQBrokerException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        SendResult sendResult = null;
        producer.start();
        int max = 100;
        for (int i = 0; i < max; i++) {
            Message msg = new Message("TopicTest",
                    "TagA",
                    "OrderID188",
                    "Hello world".getBytes(RemotingHelper.DEFAULT_CHARSET));
            sendResult = producer.send(msg);
            System.out.printf("%s%n", sendResult);
        }
        producer.setNamesrvAddr(rmqNamesrvAddr);
        producer.shutdown();
        return sendResult;
    }

    /**
     * 接收广播模式消息
     *
     */
    @Override
    public void receiveBroadcastResult( ) {

        try {
            DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("example_group_name");
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
            consumer.setNamesrvAddr(rmqNamesrvAddr);
            //set to broadcast mode
            consumer.setMessageModel(MessageModel.BROADCASTING);
            consumer.subscribe("TopicTest", "TagA || TagC || TagD");

            consumer.registerMessageListener(new MessageListenerConcurrently() {
                @Override
                public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                                ConsumeConcurrentlyContext context) {
                    System.out.printf(Thread.currentThread().getName() + " Receive New Messages: " + msgs + "%n");
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
                }
            });

            consumer.start();
            System.out.printf("Broadcast Consumer Started.%n");

        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }
}
